package com._1.message.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;


import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Collections;

@Service
@Slf4j
public class KafkaConsumerService {


//    @KafkaListener(topics = "test_2",groupId = "my-group_2")
//    public void receiveMessage(String message) {
//        System.out.println("KafkaConsumerService.listen: " + message);
//        try {
//            Thread.sleep(1000);
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//    }

    @KafkaListener(topics = "test_2", groupId = "my-group_2")
    public void receiveMessage(ConsumerRecord<String, String> record,
                               KafkaConsumer<String, String> consumer) {
        // 获取当前分区的末尾偏移量
        long endOffset = consumer.endOffsets(Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), 0L).keySet()).get(new org.apache.kafka.common.TopicPartition(record.topic(), record.partition()));
        // 计算剩余未消费的消息数量
        long remainingMessages = endOffset - record.offset() - 1;
        // 使用日志记录消息的详细信息，所有信息记录在一行
        log.info("Received message: Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}, Remaining messages: {}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value(), remainingMessages);
    }

    @KafkaListener(topics = "broadcast_topic", groupId = "consumer-group-1")
    public void receiveMessage1(String message) {
        log.info("Consumer 1 received message: {}", message);
    }

    @KafkaListener(topics = "broadcast_topic", groupId = "consumer-group-2")
    public void receiveMessage2(String message) {
        log.info("Consumer 2 received message: {}", message);
    }

    @KafkaListener(topics = "broadcast_topic", groupId = "consumer-group-3")
    public void receiveMessage4(String message) {
        log.info("Consumer 3 received message: {}", message);
    }

    @KafkaListener(topics = "random_topic", groupId = "my-random-group", id = "consumer-1")
    public void receiveMessage5(String message) {
        log.info("Consumer 1 received message: {}", message);
    }

    @KafkaListener(topics = "random_topic", groupId = "my-random-group", id = "consumer-2")
    public void receiveMessage6(String message) {
        log.info("Consumer 2 received message: {}", message);
    }

    @KafkaListener(topics = "random_topic", groupId = "my-random-group", id = "consumer-3")
    public void receiveMessage3(String message) {
        log.info("Consumer 3received message: {}", message);

    }
}
